package SparkRDD.RDD的创建

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

  /*
  Spark本身是一个并行的计算框架 --- 分区

     一、RDD是弹性分布式数据集

     二、RDD特点：
         1.RDD是数据集   --- RDD中保存了指定路径下的数据文件中的数据集
         2.RDD是一个编程模型  --- RDD变量可以调用内部的对应方法
         3.RDD相互之间有依赖关系 --- 通过不同的方法生成的RDD变量(们)之间是存在关联的
         4.RDD是可以分区的

     三、RDD的创建
         1.SparkCore的入口SparkContext

         2.RDD的创建方式：
              *通过本地数据集直接创建
              *通过读取外部数据集创建
              *通过其它RDD衍生出新RDD
   */

class wordCount {

  def main(args: Array[String]): Unit = {

    //1.创建sparkContext
    val conf = new SparkConf().setAppName("word_Count")
    val sc = new SparkContext(conf)

    // 本地运行、上传运行

    // val rdd1 = sc.textFile("dataset/wordcount.txt") // 本地运行
    // val rdd1 = sc.textFile("hdfs://192.168.64.129:9000/data/wordcount.txt")  //上传运行，提交jar包
    //./spark-submit --class cn.itcast.spark.wordCount --master spark://zyx:7077 /home/data/cn.spark-1.0-SNAPSHOT.jar

    //2.加载文件
    //   1.准备文件
    //   2.读取文件
    val rdd1:RDD[String] = sc.textFile("hdfs:///data/wordcount.txt")  //上传运行，提交jar包

    //3.处理
    //   1.拆分单词
    val rdd2:RDD[String] = rdd1.flatMap(item=>item.split(" "))

    //   2.赋予初始词频
    val rdd3:RDD[(String, Int)] = rdd2.map(item=>(item,1))

    //   3.词频聚合
    val rdd4:RDD[(String, Int)] = rdd3.reduceByKey((curr, agg)=>curr+agg)

    //   4.得到结果
    val result = rdd4.collect()
    result.foreach(item=>println(item))
  }

  @Test
  def sparkContext: Unit ={

    //1.创建sparkConf
    val conf = new SparkConf().setMaster("local[6]").setAppName("spark_context")
    //2.创建sparkContext
    val sc = new SparkContext(conf)

    //3.在使用完SparkContext后要关闭释放资源
  }


  //1.创建sparkConf
  val conf = new SparkConf().setMaster("local[6]").setAppName("spark_context")
  //2.创建sparkContext
  val sc = new SparkContext(conf)
  @Test
  //从本地集合创建RDD
  def rddCreateLocal: Unit ={
    val seq = Seq("Hello","Hi","Welcome")
    //指定数据集和分区数,以下两种方式的区别：parallelize可以不指定分区数
    val rdd1: RDD[String] = sc.parallelize(seq,2)
    val rdd2: RDD[String] = sc.makeRDD(seq,2)
  }

  @Test
  //从外部（文件）读取数据创建RDD
  def rddCreateHDFS: Unit ={
    val rdd3 = sc.textFile("hdfs:///....")

    /*
      1.textFile传入的是什么？
         * 参数传入的是一个文件的读取路径
         * hdfs：///   或  file：///
      2.该种方法创建是否支持分区？
         * 若传入的路径是 hdfs:///... 那么由HDFS中的block来决定分区
         * 注意：也可以由自己决定最小的分区数
      3.支持什么平台
         * AWS、阿里云
     */

  }

  @Test
  //从已有RDD衍生新的RDD
  def rddCreateRDD(): Unit ={
    //通过在rdd1上进行算子操作，会生成新的rdd2
    //非原地计算
    //类似于java中的str.substr 返回新的字符串且该字符串不可变
    //所以同理，新创建的RDD也是不可变的！！！
    val rdd1: RDD[Int] = sc.parallelize(Seq(1,2,3),2)
    //通过rdd1创建除了出了新的rdd2且不可变
    val rdd2: RDD[Int] = rdd1.map(item=>item)
  }

  @Test
  //map算子
  def mapTest(): Unit ={
    //  1.创建rdd
    val rdd1: RDD[Int] = sc.parallelize(Seq(1,2,3))
    //  2.执行map转换操作
    val rdd2: RDD[Int] = rdd1.map(item => item*10)
    //  3.得到结果
    val result: Array[Int] = rdd2.collect()
    //print(result)
    result.foreach(item => println(item))
  }

  @Test
  def flatMapTest(): Unit ={
    // 1.创建rdd
    val rdd1: RDD[String] = sc.parallelize(Seq("hello china","hello usa","hello uk"))
    // 2.执行flatMap转换操作
    //    * 把rdd1中的数据转换成为数组或集合的形式
    //    * 把集合或数组展开
    //    * 生成多条数据
    val rdd2: RDD[String] = rdd1.flatMap(line => line.split(" "))
    // 3.得到结果
    val result: Array[String] = rdd2.collect()
    result.foreach(item=>println(item))
    // 4.关闭sc
    sc.stop()
  }

  @Test
  def reduceByKeyTest(): Unit ={
    // 1.生成rdd1
    val rdd1: RDD[String] = sc.parallelize(Seq("hello china","hello usa","hello uk"))
//    // 2.执行flatMap转换操作
//    val rdd2: RDD[String] = rdd1.flatMap(line => line.split(" "))
//    // 3.执行map转换操作，转换为（key，values）
//    val rdd3 = rdd2.map(item=>(item,1))
//    // 3.执行reduceByKey转换操作
//    val rdd4 = rdd3.reduceByKey((curr,agg)=>curr+agg)
//    // 4.得到结果
//    val result = rdd4.collect()
//    result.foreach(item => println(item))

    // 一步生成
    val result = rdd1.flatMap(item=>item.split(" "))
                     .map(item=>(item,1))
                     .reduceByKey((curr,agg)=>curr+agg)
                     .collect()
    result.foreach(item=>println(item))
    // 5.关闭sc
    sc.stop()
  }
}
